package cn.ruc.dbiir.storm.generator.topology;

import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;

import cn.ruc.dbiir.storm.generator.bolt.DataDealBolt;
import cn.ruc.dbiir.storm.generator.spout.DataGenSpout;
import cn.ruc.dbiir.storm.generator.util.Config;

public class DataGenTopology {
	public static void main(String[] args) throws Exception {  
		
		System.out.println("params:topologyname,workersNum,tuplespersec,errorRate,spoutParalle,boltParalle,waterpath,errorpath\n");
		
		int workers = 1;
		int spoutNums = 1;
		int boltNums = 1;
		
		//提交有参数则根据参数进行配置
		if (args != null && args.length > 0) {
			Config.TOPOLOGY_NAME = args[0];
			workers = Integer.parseInt(args[1]);
			Config.TUPELS_PER_SEC = Long.parseLong(args[2]);
			Config.ERROR_RATE = Double.parseDouble(args[3]);
			spoutNums = Integer.parseInt(args[4]);
			boltNums = Integer.parseInt(args[5]);
			Config.PATH_WATER = args[6];
			Config.PATH_ERR_WATER = args[7];
		}
		
		//定义一个拓扑
		TopologyBuilder builder = new TopologyBuilder();  
		//设置一个executor（线程），默认一个
		builder.setSpout("spout", new DataGenSpout(),spoutNums);  
		//设置一个executor（线程），和一个task
		builder.setBolt("deal_bolt", new DataDealBolt(),boltNums).shuffleGrouping("spout");  
//		builder.setBolt("print", new DataPrintBolt()).shuffleGrouping("deal_bolt");  
		
		org.apache.storm.Config conf = new org.apache.storm.Config();  
		conf.setDebug(false);  
		
		
		if (args != null && args.length > 0) {  //有参数时，表示向集群提交作业，并把第一个参数当作topology的名称
			
			conf.setNumWorkers(workers);  
			StormSubmitter.submitTopology(Config.TOPOLOGY_NAME, conf, builder.createTopology());  
		} else {  //没有参数时，本地提交

			LocalCluster cluster = new LocalCluster();  
			cluster.submitTopology("test", conf, builder.createTopology());  
			Utils.sleep(80000);   //设置从启动到shutdown的间隔时间
			cluster.killTopology("test");  
			//关闭本地集群
			cluster.shutdown();  
			
		}  
	}  
}
